package com.blyat.xsoft.kernel.component.disruptor;

import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * @Auther: syh
 * @Date: 2020/7/27
 * @Description:
 */
public class PublisherBuilder<T> {
    // Publish 默认参数
    private static final int RING_SIZE = 1024 * 8;
    private static final ThreadFactory THREAD_FACTORY = Executors.defaultThreadFactory();
    private static final WaitStrategy WAIT_STRATEGY = new SleepingWaitStrategy();
    private static final ProducerType PRODUCER_TYPE = ProducerType.SINGLE;
    private static final PublishStrategy PUBLISH_STRATEGY = PublishStrategy.TRANSLATOR;

    private MetadataFactory<? extends T> metadataFactory;
    private ThreadFactory threadFactory;
    private WaitStrategy waitStrategy;
    private ErrorHandler<? extends T> errorHandler;
    private ProducerType type;
    private PublishStrategy publishStrategy;
    private MetadataPublisher<? extends T> publisher;
    private int ringSize;

    public static PublisherBuilder<Object> newBuilder() {
        return new PublisherBuilder<>();
    }

    /**
     *
     * @param ringSize 指定ringBuffer size,最好为2的n次方。默认1024*8
     * @return builder
     */
    public PublisherBuilder<T> ringSize(int ringSize) {
        this.ringSize = ringSize;
        return this;
    }

    /**
     *
     * @param metadataFactory 指定eventFactory， 默认EventFactory
     * @return builder
     */
    public <T1 extends T> PublisherBuilder<T> metadataFactory(MetadataFactory<T1> metadataFactory) {
        this.metadataFactory = metadataFactory;
        return this;
    }

    /**
     *
     * @param threadFactory 指定ThreadFactory, 默认Executors.defaultThreadFactory()
     * @return builder
     */
    public PublisherBuilder<T> threadFactory(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
        return this;
    }

    /**
     *
     * @param waitStrategy 指定等待策略, 默认SleepingWaitStrategy
     * @return builder
     */
    public PublisherBuilder<T> waitStrategy(WaitStrategy waitStrategy) {
        this.waitStrategy = waitStrategy;
        return this;
    }

    /**
     *
     * @param publishStrategy 发布策略
     * @return builder
     */
    public PublisherBuilder<T> publishStrategy(PublishStrategy publishStrategy) {
        this.publishStrategy = publishStrategy;
        return this;
    }

    /**
     *
     * @param type 指定producerType
     * @return builder
     */
    public PublisherBuilder<T> producerType(ProducerType type) {
        this.type = type;
        return this;
    }

    /**
     *
     * @param errorHandler 异常处理类
     * @return builder
     */
    public <T1 extends T> PublisherBuilder<T> errorHandler(ErrorHandler<T1> errorHandler) {
        this.errorHandler = errorHandler;
        return this;
    }

    /**
     *
     * 初始化disruptor
     * @return builder
     */
    public <T1 extends T> PublisherBuilder<T> disruptor() {
        this.metadataFactory = this.metadataFactory == null ? new MetadataFactory<T1>() : this.metadataFactory;
        this.threadFactory = this.threadFactory == null ? THREAD_FACTORY : this.threadFactory;
        this.waitStrategy = this.waitStrategy == null ? WAIT_STRATEGY : this.waitStrategy;
        this.ringSize = this.ringSize <= 0 ? RING_SIZE : this.ringSize;
        this.type = this.type == null ? PRODUCER_TYPE : this.type;
        this.publishStrategy = this.publishStrategy == null ? PUBLISH_STRATEGY : this.publishStrategy;

        @SuppressWarnings("unchecked")
        PublisherBuilder<T1> builder = (PublisherBuilder<T1>) this;
        this.publisher = new MetadataPublisher<>(builder);
        return this;
    }

    /**
     *
     * @param producers 配置生产者
     * @return builder
     */
    @SafeVarargs
    @SuppressWarnings("unchecked")
    public final <T1 extends T> PublisherBuilder<T> producer(MetadataProducer<T1>... producers) {
        if (isInit()) {
            ((MetadataPublisher<T1>) this.publisher).producer(producers);
        }
        return this;
    }

    /**
     *
     * @param eventHandlers 每个event事件可以被所有handler处理
     * @param <T1> 泛型
     * @return builder
     */
    @SafeVarargs
    @SuppressWarnings("unchecked")
    public final <T1 extends T> PublisherBuilder<T> handler(Handler<T1>... eventHandlers) {
        if (isInit()) {
            ((MetadataPublisher<T1>) this.publisher).eventHandler(eventHandlers);
        }
        return this;
    }

    /**
     *
     * @param workHandlers 每个event事件只能被一个work处理
     * @param <T1> 泛型
     * @return builder
     */
    @SafeVarargs
    @SuppressWarnings("unchecked")
    public final <T1 extends T> PublisherBuilder<T> worker(Handler<T1>... workHandlers) {
        if (isInit()) {
            ((MetadataPublisher<T1>) this.publisher).workHandler(workHandlers);
        }
        return this;
    }

    /**
     *
     * @param thenEventHandlers handler或work之后处理
     * @param <T1> 泛型
     * @return builder
     */
    @SafeVarargs
    @SuppressWarnings("unchecked")
    public final <T1 extends T> PublisherBuilder<T> then(Handler<T1>... thenEventHandlers) {
        if (isInit()) {
            ((MetadataPublisher<T1>) this.publisher).thenHandler(thenEventHandlers);
        }
        return this;
    }

    /**
     *
     * @param afterEventHandlers 顺序指定handler，不可接直接在worker方法后面调用
     * @param <T1> 泛型
     * @return builder
     */
    @SafeVarargs
    @SuppressWarnings("unchecked")
    public final <T1 extends T> PublisherBuilder<T> after(Handler<T1>... afterEventHandlers) {
        if (isInit()) {
            ((MetadataPublisher<T1>) this.publisher).afterHandler(afterEventHandlers);
        }
        return this;
    }

    private boolean isInit() {
        if (this.publisher == null) {
            throw new IllegalStateException("execute disruptor() function before set handlers or workers.");
        }
        return true;
    }

    @SuppressWarnings("unchecked")
    public <T1 extends T> Publisher<T1> build() {
        return ((MetadataPublisher<T1>) this.publisher).start();
    }

    @SuppressWarnings("unchecked")
    public <T1 extends T> MetadataFactory<T1> getMetadataFactory() {
        return (MetadataFactory<T1>) metadataFactory;
    }

    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }

    public WaitStrategy getWaitStrategy() {
        return waitStrategy;
    }

    @SuppressWarnings("unchecked")
    public <T1 extends T> ErrorHandler<T1> getErrorHandler() {
        return (ErrorHandler<T1>) errorHandler;
    }

    public ProducerType getType() {
        return type;
    }

    public PublishStrategy getPublishStrategy() {
        return publishStrategy;
    }

    public int getRingSize() {
        return ringSize;
    }
}
